package cn.ccydrools;

import cn.ccydrools.pojo.a03Event;
import org.apache.commons.io.FileUtils;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.common.state.MapState;
import org.apache.flink.api.common.state.MapStateDescriptor;
import org.apache.flink.api.common.typeinfo.TypeHint;
import org.apache.flink.api.common.typeinfo.TypeInformation;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.api.java.tuple.Tuple3;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.KeyedStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.KeyedProcessFunction;
import org.apache.flink.util.Collector;
import org.kie.api.io.ResourceType;
import org.kie.api.runtime.KieSession;
import org.kie.internal.utils.KieHelper;

import java.io.File;
import java.nio.charset.StandardCharsets;

public class _04FlinkDroolsDemo {

    public static void main(String[] args) throws Exception {

        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        //用户，商品分类ID，事件类型
        //u1001,c201,view
        //一个用户，在一天之内，浏览了某个分类的商品次数大于2就触发一次事件
        DataStreamSource<String> lines = env.socketTextStream("localhost", 8888);

        SingleOutputStreamOperator<Tuple3<String, String, String>> tpStream = lines.map(new MapFunction<String, Tuple3<String, String, String>>() {
            @Override
            public Tuple3<String, String, String> map(String value) throws Exception {
                String[] fields = value.split(",");
                return Tuple3.of(fields[0], fields[1], fields[2]);
            }
        });

        //将同一用户的行为数据搞到一起
        KeyedStream<Tuple3<String, String, String>, String> keyed = tpStream.keyBy(t -> t.f0);

        keyed.process(new KeyedProcessFunction<String, Tuple3<String, String, String>, Tuple3<String, String, String>>() {

            private transient MapState<Tuple2<String, String>, Integer> mapState;

            private transient KieSession kieSession;

            @Override
            public void open(Configuration parameters) throws Exception {
                //定义一个状态描述器
                //((c201,view), 3)
                MapStateDescriptor<Tuple2<String, String>, Integer> stateDescriptor = new MapStateDescriptor<>("category-event-count", TypeInformation.of(new TypeHint<Tuple2<String, String>>() {
                }), TypeInformation.of(new TypeHint<Integer>() {}));

                mapState = getRuntimeContext().getMapState(stateDescriptor);

                KieHelper kieHelper = new KieHelper();
                //将drools的规则文件转成字符串
                String droolsStr = FileUtils.readFileToString(new File("rules/first-demo.drl"), StandardCharsets.UTF_8);

                //添加规则
                kieHelper.addContent(droolsStr, ResourceType.DRL);

                //创建规则匹配的会话(将用户传入的数据和事先添加的规则进行匹配)
                kieSession = kieHelper.build().newKieSession();

            }

            @Override
            public void processElement(Tuple3<String, String, String> value, Context ctx, Collector<Tuple3<String, String, String>> out) throws Exception {

                String uid = value.f0; //u1001
                String cid = value.f1; //c201
                String type = value.f2; //view

                Tuple2<String, String> key = Tuple2.of(cid, type);
                Integer count = mapState.get(key);
                if(count == null) {
                    count = 0;
                }
                count++;
                mapState.put(key, count);

                a03Event event = new a03Event(type, count, false);

                kieSession.insert(event);
                kieSession.fireAllRules();

                if (event.isHit()) {
                    //System.out.println("事件满足了！");
                    out.collect(Tuple3.of(uid, "发优惠卷", "满30减5"));
                }
            }
        }).print();


        env.execute();
    }
}
